package com.itc.bi.kafaka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.itc.bi.entity.*;
import com.itc.bi.handler.dataSet.Database;
import com.itc.bi.kafaka.DTO.MqttFieldDto;
import com.itc.bi.service.DataSetColumnService;
import com.itc.bi.service.DirectDataSetService;
import com.itc.bi.service.DirectoryService;
import com.itc.bi.service.MqMqttClientService;
import com.itc.bi.service.impl.ClickHouseJDBCService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * @program: itc-airport-cloud
 * @description: mqtt数据处理
 * @author: songxiude
 * @create: 2024-07-17 11:20
 **/
@Service
@Slf4j
public class MqttDataAcceptService {

    @Autowired
    private MqMqttClientService mqttClientService;

    @Autowired
    private DirectoryService directoryService;

    @Autowired
    private DirectDataSetService directDataSetService;

    @Autowired
    private DataSetColumnService dataSetColumnService;

    @Autowired
    private ClickHouseJDBCService jdbcService;

    @Autowired
    private Database database;

    /**
     * 处理mqtt数据入库
     *
     * @param str
     */
    @Async("asyncServiceExecutor")
    public void dataAccept(String str) {
        JSONObject jsonObject = JSONObject.parseObject(str, Feature.OrderedField);
        String topic = jsonObject.getString("topic");
        String clientId = jsonObject.getString("clientId");
        //1、判断表是否已经存在，如果不存在先创建数据集(按百度mqtt接入规范)
        String[] split = topic.split("/");
        //根据第二段第三段作为数据表名
        String tablename = "mqtt_" + split[1] + "_" + split[2];
        long count = directoryService.count(Wrappers.<DirectoryEntity>lambdaQuery().eq(DirectoryEntity::getTableName, tablename));
        MqMqttClientEntity mqMqttClient = mqttClientService.getOne(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, clientId));
        log.info("设备名称加设备编号：" + tablename);
        JSONObject data = jsonObject.getJSONObject("data");

        //2、满足条件创建数据集
        if (count == 0) {
            List<MqttFieldDto> fields = this.getFields(data);
            //1、回填tb_direct_data_set
            DirectDataSetEntity directDataSet = new DirectDataSetEntity();
            //生成id回填
            long ddsid = IdWorker.getId();
            directDataSet.setId(ddsid);
            directDataSet.setDtId(mqMqttClient.getDtId());
            directDataSet.setTableName(tablename);
            directDataSet.setType("MQTT");
            directDataSetService.save(directDataSet);
            //2、回填tb_directory
            DirectoryEntity directoryEntity = new DirectoryEntity();
            directoryEntity.setDtId(ddsid);
            directoryEntity.setPid(mqMqttClient.getDtId());
            directoryEntity.setTableName(tablename);
            directoryEntity.setIsDir(1);
            directoryService.save(directoryEntity);
            //3、回填tb_data_set_column
            boolean datasetB = this.saveDataSetColumn(ddsid, fields);
            System.out.println("创建字段结果：" + datasetB);
            //4、先创建目标表
            jdbcService.createTableSourceDS(tablename, ddsid, "");

            log.info("已经生成目标表：" + tablename);

        }
        //3、插入数据
        List<Map<String, Object>> maps = this.createData(data);
        jdbcService.insertDataByTableHashMap(tablename, maps);


    }

    /**
     * 创建datasetcolumn表数据
     *
     * @param dsetId
     * @param fields
     * @return
     */
    public boolean saveDataSetColumn(Long dsetId, List<MqttFieldDto> fields) {
        List<DataSetColumnEntity> columns = new ArrayList<>();
        for (int i = 0; i < fields.size(); i++) {
            MqttFieldDto mqttFieldDto = fields.get(i);
            DataSetColumnEntity dataSetColumnEntity = new DataSetColumnEntity();
            dataSetColumnEntity.setDataSetId(dsetId);
            dataSetColumnEntity.setColumnName(mqttFieldDto.getField());
            dataSetColumnEntity.setColumnComment(mqttFieldDto.getField());
            dataSetColumnEntity.setColumnPosition(i + 1);
            dataSetColumnEntity.setColumnType(mqttFieldDto.getFieldType());
            columns.add(dataSetColumnEntity);
        }
        return dataSetColumnService.saveBatch(columns);
    }

    /**
     * 获取字段列表
     *
     * @param data
     * @return
     */
    public List<MqttFieldDto> getFields(JSONObject data) {
        List<MqttFieldDto> mqttFieldDtos = new ArrayList<>();
        for (String key : data.keySet()) {
            MqttFieldDto mqttFieldDto = new MqttFieldDto();
            Object o = data.get(key);
            if (o instanceof JSONObject) {
                JSONObject jsonObject1 = data.getJSONObject(key);
                if (this.countLevels(jsonObject1) > 1) {
                    for (String lfied : jsonObject1.keySet()) {
                        Object lo = jsonObject1.get(lfied);
                        if (lo instanceof JSONObject) {
                            JSONObject jsonObject2 = jsonObject1.getJSONObject(lfied);
                            for (String rfield : jsonObject2.keySet()) {
                                MqttFieldDto mqttFieldB = new MqttFieldDto();
                                mqttFieldB.setField(key + "_" + lfied + "_" + rfield);
                                mqttFieldB.setFieldType("double");
                                mqttFieldDtos.add(mqttFieldB);
                            }
                        } else {
                            MqttFieldDto mqttField = new MqttFieldDto();
                            mqttField.setField(key + "_" + lfied);
                            mqttField.setFieldType("double");
                            mqttFieldDtos.add(mqttField);
                        }

                    }
                }
            } else if (o instanceof JSONArray) {
                JSONArray jsonArray = data.getJSONArray(key);
                JSONObject json = (JSONObject) jsonArray.get(0);
                for (String ks : json.keySet()) {
                    Object aobj = json.get(ks);
                    if (aobj instanceof JSONObject) {
                        JSONObject laobj = json.getJSONObject(ks);
                        for (String ksl : laobj.keySet()) {
                            if (laobj.get(ksl) instanceof JSONObject) {
                                JSONObject raobj = laobj.getJSONObject(ksl);
                                for (String ksr : raobj.keySet()) {
                                    MqttFieldDto mqttFieldB = new MqttFieldDto();
                                    mqttFieldB.setField(ks + "_" + ksl + "_" + ksr);
                                    mqttFieldB.setFieldType("double");
                                    mqttFieldDtos.add(mqttFieldB);
                                }
                            }
                        }
                    } else {
                        MqttFieldDto mqttField = new MqttFieldDto();
                        if (ks.equals("timestamp")) {
                            mqttField.setField(ks + "_ts");
                        } else {
                            mqttField.setField(ks);
                        }
                        mqttField.setFieldType("double");
                        mqttFieldDtos.add(mqttField);
                    }
                }

            } else if (o instanceof String) {
                mqttFieldDto.setField(key);
                mqttFieldDto.setFieldType("string");
            } else {
                String tmkey = null;
                if (key.equals("timestamp")) {
                    tmkey = key + "_time";
                    mqttFieldDto.setField(key + "_time");
                } else {
                    tmkey = key;
                    mqttFieldDto.setField(key);
                }
                mqttFieldDto.setFieldType("double");
            }
            mqttFieldDtos.add(mqttFieldDto);

            //如果是数组对象再组装数据集

        }
        //最后一个else没有判断类型导致多出一个空对象后续观察是否要严格判断类型
        mqttFieldDtos.remove(mqttFieldDtos.size() - 1);
        String fields = JSON.toJSONString(mqttFieldDtos);
        log.info("组装字段：" + fields);

        return mqttFieldDtos;
    }

    /**
     * 组装数据字段
     *
     * @param data
     * @return
     */
    public List<Map<String, Object>> createData(JSONObject data) {
        List<Map<String, Object>> maps = new ArrayList<>();
        JSONArray array = null;
        Map<String, Object> map = new LinkedHashMap<>();
        for (String key : data.keySet()) {
            Object o = data.get(key);
            if (o instanceof JSONObject) {
                JSONObject jsonObject1 = data.getJSONObject(key);
                if (this.countLevels(jsonObject1) > 1) {
                    for (String lfied : jsonObject1.keySet()) {
                        Object lo = jsonObject1.get(lfied);
                        if (lo instanceof JSONObject) {
                            JSONObject jsonObject2 = jsonObject1.getJSONObject(lfied);
                            for (String rfield : jsonObject2.keySet()) {
                                map.put(key + "_" + lfied + "_" + rfield, jsonObject2.get(rfield));
                            }
                        } else {
                            map.put(key + "_" + lfied, lo);
                        }

                    }
                }
            } else if (o instanceof String) {
                map.put(key, o);

            } else if (o instanceof Long || o instanceof Integer) {
                String tmkey = null;
                if (key.equals("timestamp")) {
                    tmkey = key + "_time";
                } else {
                    tmkey = key;
                }

                map.put(tmkey, o);

            }
            //如果是数组对象赋值给空对象往下处理
            if (o instanceof JSONArray) {
                array = data.getJSONArray(key);
            }

        }
        //如果是对象数组重新组装数据
        if (CollectionUtils.isEmpty(array)) {
            maps.add(map);
        } else {
            for (int i = 0; i < array.size(); i++) {
                Map<String, Object> mapnew = new LinkedHashMap();
                for (String s : map.keySet()) {
                    mapnew.put(s, map.get(s));
                }
                JSONObject json = (JSONObject) array.get(i);
                for (String j : json.keySet()) {
                    Object jk = json.get(j);
                    if (jk instanceof JSONObject) {
                        JSONObject object = json.getJSONObject(j);
                        for (String k : object.keySet()) {
                            JSONObject jsond = object.getJSONObject(k);
                            for (String l : jsond.keySet()) {
                                //Map<String,Object> lmap=new HashMap<>();
                                mapnew.put(j + "_" + k + "_" + l, jsond.get(l));

                            }
                        }
                    } else {
                        String ksy = null;
                        if (j.equals("timestamp")) {
                            ksy = j + "_ts";
                        } else {
                            ksy = j;
                        }
                        mapnew.put(ksy, jk);

                    }

                }
                maps.add(mapnew);
            }
        }
        String s = JSON.toJSONString(maps);
        log.info("组装的数据：" + s);

        return maps;
    }

    /**
     * 获取json有多少个层级
     *
     * @param jsonObject
     * @return
     */
    public int countLevels(JSONObject jsonObject) {
        int level = 1;
        for (String key : jsonObject.keySet()) {
            Object value = jsonObject.get(key);
            if (value instanceof JSONObject) {
                level += countLevels((JSONObject) value);
            }
        }
        return level;
    }
}